﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.ComplexEventProcessing;
using Microsoft.ComplexEventProcessing.ManagementService;
using Microsoft.ComplexEventProcessing.Linq;
using Microsoft.Samples.SmartGrid.AMI.MeterEventType;
using Microsoft.Samples.SmartGrid.AMI.WcfInputAdapter;
using Microsoft.Samples.SmartGrid.AMI.ServiceBrokerOutputAdapter;
using Microsoft.Samples.SmartGrid.AMI.WcfOutputAdapter;
using System.ServiceModel;
using System.Configuration;

namespace Microsoft.Samples.SmartGrid.AMI.SIHost
{
    class Program
    {
        static void Main(string[] args)
        {
            #region configuration
            string instanceName = ConfigurationManager.AppSettings["InstanceName"].ToString();
            string managementServiceEndpoint = ConfigurationManager.AppSettings["ManagementServiceEndpoint"].ToString();
            string inputAdapterEndpoint = ConfigurationManager.AppSettings["InputAdapterEndpoint"].ToString();
            string applicationName = ConfigurationManager.AppSettings["ApplicationName"].ToString();
            string streamName = ConfigurationManager.AppSettings["StreamName"].ToString();
            string serviceBusEndpoint = ConfigurationManager.AppSettings["ServiceBusEndpoint"].ToString();
            string sbConnString = ConfigurationManager.AppSettings["SBConnString"].ToString();
            string sbMessageType = ConfigurationManager.AppSettings["SBMessageType"].ToString();
            string sbServiceContract = ConfigurationManager.AppSettings["SBServiceContract"].ToString();
            string sbServiceName = ConfigurationManager.AppSettings["SBServiceName"].ToString();
            #endregion

            //Create an embeded instance of StreamInsight
            Server cepServer = Server.Create(instanceName); 

            using (cepServer)
            {

                #region Setup Management Service

                //Create the management service so we can use 
                //Event Flow Debugger
                var mservice = cepServer.CreateManagementService();
                ServiceHost cepManagementHost = new ServiceHost(mservice);
                cepManagementHost.AddServiceEndpoint(
                    typeof(IManagementService), 
                    new WSHttpBinding(SecurityMode.Message),
                    managementServiceEndpoint);
                cepManagementHost.Open();

                #endregion

                //Create the StreamInsight Application
                Application cepApplication = cepServer.CreateApplication(applicationName);

                #region InputStream Configuration

                //Create the InputStream and configure the adapter
                //  --give the stream a name
                //  --pass the adapter factory
                //  --pass the config
                //  --set the shape
                var meterInputStream = CepStream<RawMeterEventType>.Create(
                    streamName, 
                    typeof(WcfInputAdapterFactory),
                    new WcfInputAdapterConfig() 
                        { 
                            CtiFrequency = 1,
                            ExplicitCtiPublishingFrequency=10000,
                            BaseAddress = new Uri(inputAdapterEndpoint) 
                        }, 
                    EventShape.Point);

            
                //Publish the stream so it can be used by other queries
                //  --set the URI
                //  --the application
                //  --name
                //  --description
                //  --shape
                //  --ordering
                var publishedRawMeterReadings = meterInputStream.ToPublishedStreamQuery(
                    new Uri("cep:/Server/Application/" + applicationName + "/PublishedStream/" + streamName),
                    cepApplication,
                    "RawEvents",
                    "All Meter Events",
                    EventShape.Point,
                    StreamEventOrder.FullyOrdered);
                
                #endregion

                #region RawEvents To ServiceBroker

                //Consume the published stream
                //  --set the URI
                //  --set the event shape
                var sbEventStream = CepStream<RawMeterEventType>.Create(
                    new Uri("cep:/Server/Application/"+ applicationName +"/PublishedStream/" + streamName), 
                    EventShape.Point);

                //Define a passthrough query
                var sbPassthrough = from e in sbEventStream select e;

                //Bind the inputstream, query, and output adapter together
                //  --application name
                //  --query name
                //  --description
                //  --pass the adapter factory
                //  --pass the adapter config
                //  --event shape
                //  --order
                var sbQuery = sbPassthrough.ToQuery(
                    cepApplication,
                    "QueryForServiceBroker", 
                    "Passthrough query to send raw events to service broker service", 
                    typeof(ServiceBrokerOutputAdapterFactory), 
                    new ServiceBrokerOutputAdapterConfig 
                        {
                            connString = sbConnString,
                            messageType=sbMessageType,
                            serviceContract=sbServiceContract,
                            serviceName = sbServiceName
                        }, 
                    EventShape.Point, 
                    StreamEventOrder.FullyOrdered);

                #endregion

                #region Filtered Events to BizTalk

                var btsEventStream = CepStream<RawMeterEventType>.Create(new Uri("cep:/Server/Application/" + applicationName + "/PublishedStream/" + streamName), EventShape.Point);

                //Config Timout
                var timeout = TimeSpan.FromSeconds(15);

                //Get valid events for each meter
                var btsValidOutages = from alarm in btsEventStream.Where(e => e.EventType == "O")
                                      where (from nextevent in btsEventStream
                                          .AlterEventLifetime(
                                              e => e.StartTime.Subtract(timeout),
                                              e => timeout)
                                             where (nextevent.EventType == "T" && nextevent.MeterNumber == alarm.MeterNumber)
                                             select nextevent).IsEmpty()

                                      select alarm;

                // Expand all alarm events to the timeout and set the Index of the event in the snapshot
                var alarmsInWindow = from meter in btsValidOutages
                                     group meter by meter.MeterNumber into meterGroup

                                     from window in meterGroup
                                      .Where(e => e.EventType == "O")
                                      .AlterEventDuration(e => timeout)
                                      .SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
                                     select new { Index = window.Count() };

                // Those alarmsInWindow entries with a Index of 1 are the intial alarm
                // convert to a point and join with original stream.
                var btsDistinctOutages = from c in alarmsInWindow
                                .Where(e => e.Index == 1)
                                .ToPointEventStream()
                                         from e in btsValidOutages
                                         select e;

                var btsQuery = btsDistinctOutages.ToQuery(
                    cepApplication,
                    "QueryForBizTalk",
                    "Passthrough query to send raw events to service BizTalk",
                    typeof(WcfOutputAdapterFactory),
                    new WcfOutputAdapterConfig
                    {
                        endpointAddress = serviceBusEndpoint
                    },
                    EventShape.Point,
                    StreamEventOrder.FullyOrdered);

                #endregion

                publishedRawMeterReadings.Start();
                sbQuery.Start();
                btsQuery.Start();

                Console.WriteLine("StreamInsight Host Started....");
                Console.WriteLine("press <enter> to shut down.");
                Console.ReadLine();

                btsQuery.Stop();
                sbQuery.Stop();
                publishedRawMeterReadings.Stop();
            }
        }
    }
}
